package xin.nick.roc.consumer.client;

import lombok.extern.slf4j.Slf4j;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.*;
import org.apache.curator.retry.ExponentialBackoffRetry;


import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author Nick
 * @date 2021/7/11
 * @description
 */
@Slf4j
public class ZkRegister {

    private static CuratorFramework client;

    public static void start(String zk) {
        try {
            String path = "/provider";
            RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);

            client = CuratorFrameworkFactory.builder()
                    .connectString(zk)
                    .sessionTimeoutMs(50000)
                    .connectionTimeoutMs(30000)
                    .retryPolicy(retryPolicy)
                    .namespace("m-rpc")
                    .build();
            client.start();
            log.info("zk会话创建");

            // 准备监听事件
            CuratorCache curatorCache = CuratorCache.build(client, path);
            curatorCache.listenable().addListener(new CuratorCacheListener() {
                @Override
                public void event(Type type, ChildData childData, ChildData childData1) {
                    if(Type.NODE_CREATED.equals(type)
                        || Type.NODE_DELETED.equals(type)) {
                        // 如果发生增加或减少提供者的事件,则进行更新提供者列表
                        log.info("提供者发生变化,刷新负载均衡列表");
                        try {
                            List<String> providerList = client.getChildren().forPath(path);
                            if (providerList.size() == 0) {
                                return;
                            }
                            // 拿到原来所有的rpc对象,
                            List<RpcClient> oldRpcClientList = new ArrayList<>();
                            for(RpcClient rpcClient : ProviderInfo.getRpcClientMap().values()) {
                                oldRpcClientList.add(rpcClient);

                                // 同时清除掉响应时间信息
                            }

                            // 准备新的rpc对象
                            Map<String, RpcClient> rpcClientMap = new HashMap<>();
                            List<RpcClient> newRpcClientList = new ArrayList<>();
                            for(String providerString : providerList) {
                                String[] providerInfoArray = providerString.split(":");
                                String ip = providerInfoArray[0];
                                String port = providerInfoArray[1];
                                // 拿到信息之后,创建链接
                                rpcClientMap.put(providerString, new RpcClient(ip, Integer.valueOf(port)));
                            }

                            // 更新所有的rpc对象
                            ProviderInfo.setRpcClientMap(rpcClientMap);

                            // 再将原来的rpc结束掉
                            for (RpcClient rpcClient : oldRpcClientList) {
                                rpcClient.close();
                            }
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                }
            });

            curatorCache.start();

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static CuratorFramework getClient() {
        return client;
    }

}
